package com.lrd.rmqcluster.rmq5.testproducer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Component
public class TestMessageProducer {

    private static final Logger log = LoggerFactory.getLogger(TestMessageProducer.class);

    private DefaultMQProducer producer;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    @Value("${apache.rocketmq.test.topic}")
    private String topic;
    @Value("${apache.rocketmq.test.group}")
    private String group;

    @PostConstruct
    public void start() {
        producer = new DefaultMQProducer(group);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setInstanceName(topic);
        try {
            log.info("================================init testProducer================================");
            producer.start();
            log.info("================================init testProducer success!!!================================");
        } catch (MQClientException e) {
            log.error("init testProducer生产者启动失败,IP地址:{}===group:{}", namesrvAddr, group, e);
            throw new RuntimeException("init POINT_ITEM MQ producer fail", e);
        }
    }

    public boolean sendMsg(String message, String key) {
        log.info("testProducer发送消息|message={}", message);
        try {
            byte[] messageData = message.getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message mqMsg = new Message(topic, "tag", key, messageData);
            SendResult sendResult = producer.send(mqMsg);
            return sendResult != null && sendResult.getSendStatus() == SendStatus.SEND_OK;
        } catch (Exception e) {
            log.error("testProducer发送消息异常", e);
            return false;
        }
    }


    @PreDestroy
    public void stop() {
        producer.shutdown();
        log.info("================================shutdown testProducer================================");
    }
}
